#include "cgzCThreadPool.h"
#include "cgzCSpdLog.h"

cgzCThread::cgzCThread(cgzCThreadPool *pThreadPool)
    : m_pTask(nullptr)
{
    m_bRunning = false;
    m_pThreadPool = pThreadPool;
}

cgzCThread::~cgzCThread()
{
    stop();
    m_pThreadPool = nullptr;
    m_bRunning = false;
}

void cgzCThread::start()
{
    if (!m_bRunning)
    {
        m_bRunning = true;
        m_thread = new std::thread(&cgzCThread::run, this);
        m_pThreadPool->push_thread(m_thread);
    }
}

void cgzCThread::stop()
{
    if (m_bRunning)
    {
        m_bRunning = false;
    }
}

void cgzCThread::wakeUp()
{
    cv.notify_one();
}

bool cgzCThread::setbusy(cgzITask *pTask)
{
    std::unique_lock lock(mtx_opt);
    m_pTask = pTask;
    return true;
}

bool cgzCThread::setIdle()
{
    std::unique_lock lock(mtx_opt);
    m_pTask->setFree();
    m_pTask = nullptr;
    return true;
}

bool cgzCThread::hasIdle()
{
    std::shared_lock lock(mtx_opt);
    return m_pTask == nullptr;
}

thread_local int threadTaskId = -1;
void cgzCThread::run()
{
    while (m_bRunning)
    {
        if (m_pTask != nullptr)
        {
            threadTaskId = m_pTask->getId();
            auto start = fnowmls();
            // 任务执行
            bool ret = m_pTask->run(start);
            auto close = fnowmls();
            auto pms = close - start;
            if (pms > 15)
            {
                auto that = m_pTask;
                cgz_that_error("{} cost: {} ms", m_pTask->getName(), pms);
            }
            // 任务执行时间过长
            if (ret)
            {
                // 设置空闲等待唤醒
                setIdle();
                m_pThreadPool->wakeUp();
            }
        }
        else
        {
            auto timeout = std::chrono::steady_clock::now() + std::chrono::milliseconds(2);
            std::unique_lock<std::mutex> run_lock(mtx_run);
            cv.wait_until(run_lock, timeout, [this]
                          { return m_pTask != nullptr || !m_bRunning; });
        }
    }
}

cgzCThreadPool *cgzCThreadPool::m_pInstance = nullptr;
cgzCThreadPool *cgzCThreadPool::Instance(int nThreadNum)
{
    if (nThreadNum <= 0)
    {
        return m_pInstance;
    }
    if (nullptr == m_pInstance)
    {
        m_pInstance = new cgzCThreadPool(nThreadNum);
    }

    return m_pInstance;
}

cgzCThreadPool::cgzCThreadPool(int nThreadNum)
    : m_thread(nullptr)
{
    m_nThreadNum = nThreadNum;
    m_pThreadList = new cgzCThread *[m_nThreadNum];
    for (int i = 0; i < m_nThreadNum; i++)
    {
        m_pThreadList[i] = new cgzCThread(this);
    }
}

cgzCThreadPool::~cgzCThreadPool()
{
    for (int i = 0; i < m_nThreadNum; i++)
    {
        delete m_pThreadList[i];
    }
    delete[] m_pThreadList;
}

void cgzCThreadPool::start()
{
    for (int i = 0; i < m_nThreadNum; i++)
    {
        m_pThreadList[i]->start();
    }
    if (nullptr == m_thread)
    {
        m_thread = new std::thread(&cgzCThreadPool::run, this);
        push_thread(m_thread);
    }
}

void cgzCThreadPool::stop()
{
    std::shared_lock lock(mtx_opt);

    for (int i = 0; i < m_nThreadNum; i++)
    {
        m_pThreadList[i]->stop();
    }

    for (auto pth : m_threadAll)
    {
        pth->join();
    }
    m_bRunning = false;
}

// 注册服务本身
void cgzCThreadPool::addTask(const char *origin, cgzITask *pTask)
{
    std::unique_lock lock(mtx_que);
    auto it = m_taskNMap.find(origin);
    if (it != m_taskNMap.end())
    {
        assert(false);
    }

    m_taskMap.insert(std::make_pair(pTask->getId(), pTask));
    m_taskNMap.insert(std::make_pair(origin, pTask));
    m_taskINMap.insert(std::make_pair(pTask->getId(), origin));
    m_taskIdleMap.insert(std::make_pair(pTask->getId(), pTask));
    if (!pTask->basBusy())
    {
        m_taskIdleQueue.push_back(pTask);
    }
    else
    {
        m_taskBusyQueue.push_back(pTask);
    }
}

// 添加消费后返回任务|添加自产生任务
void cgzCThreadPool::addTask(const char *origin, const char *target, cgzIData *pTask)
{
    std::shared_lock lock(mtx_que);
    auto oit = m_taskNMap.find(origin);
    auto tit = m_taskNMap.find(target);
    pTask->origin = oit->second->getId();
    pTask->target = tit->second->getId();
    oit->second->retTask(pTask);
}

// 添加消费后返回任务|添加自产生任务
void cgzCThreadPool::addTask(const int origin, const int target, cgzIData *pTask)
{
    pTask->origin = origin;
    pTask->target = target;
    auto oit = m_taskMap.find(origin);
    oit->second->retTask(pTask);
}

void cgzCThreadPool::setLog(int logtid, int logfid, int loglvl)
{
    if (-1 == m_logtid || -1 == m_logfid || -1 == m_loglvl)
    {
        // 只记录第一个
        m_logtid = logtid;
        m_logfid = logfid;
        m_loglvl = loglvl;
    }
}

void cgzCThreadPool::errLog(const char *file, const char *func, const int line, const char *msg)
{
    std::string msgstr = fmt::format("{}:{} {}:{}", file, line, func, msg).c_str();
    if (-1 == m_logfid || -1 == m_loglvl || -1 == m_logtid)
    {
        // 未设置日志
        std::cerr << msgstr << std::endl;
        return;
    }
    else if (-1 == threadTaskId)
    {
        // 非任务线程
        std::cerr << msgstr << std::endl;
        return;
    }

    // 日志写入
    auto node = malloc(sizeof(cgzCSpdItem));
    auto item = new (node) cgzCSpdItem(m_logfid, m_loglvl, msgstr);
    addTask(threadTaskId, m_logtid, item);
}

const char *cgzCThreadPool::getname(int origin)
{
    auto oit = m_taskINMap.find(origin);
    return oit->second.c_str();
}

const cgzITask *cgzCThreadPool::getTask(int id)
{
    auto it = m_taskMap.find(id);
    return it->second;
}

int cgzCThreadPool::getidbyname(const char *name)
{
    std::shared_lock lock(mtx_que);
    auto it = m_taskNMap.find(name);
    return it->second->getId();
}

void cgzCThreadPool::push_thread(std::thread *pth)
{
    std::unique_lock lock(mtx_opt);
    m_threadAll.push_back(pth);
}

void cgzCThreadPool::wakeUp()
{
    m_bwake = true;
    cv.notify_one();
}

void cgzCThreadPool::wakeDown()
{
    m_bwake = false;
}

bool cgzCThreadPool::isrunning()
{
    return m_bRunning;
}

void cgzCThreadPool::shutdown()
{
    m_bShutdown = true;
}

void cgzCThreadPool::waits()
{
    while (m_bRunning)
    {
        std::this_thread::sleep_for(std::chrono::milliseconds(100));
    }
    stop();
}

void cgzCThreadPool::run()
{
    while (m_bRunning)
    {
        exec();
    }
}

void cgzCThreadPool::exec()
{
    // 等待唤醒
    auto timeout = std::chrono::steady_clock::now() + std::chrono::milliseconds(2);
    std::unique_lock<std::mutex> run_lock(mtx_run);
    cv.wait_until(run_lock, timeout, [this]
                  { return m_bwake || !m_bRunning; });
    wakeDown();

    // 取出完成任务
    std::shared_lock lock(mtx_que);
    auto nms = fnowmls();
    // 处理自产事件-变成繁忙任务
    for (auto it = m_taskIdleQueue.begin(); it != m_taskIdleQueue.end();)
    {
        auto task = *it;
        if (task->casBusy())
        {
            m_taskIdleMap.erase(task->getId()); // 从 map 中移除

            // 将节点从 m_taskIdleQueue 直接转移到 m_taskBusyQueue
            auto next_it = std::next(it); // 先保存下一个迭代器（因为 splice 会修改 it）
            m_taskBusyQueue.splice(m_taskBusyQueue.end(), m_taskIdleQueue, it);
            it = next_it; // 继续遍历
        }
        else
        {
            it++;
        }
    }

    // 处理繁忙任务队列
    while (!m_taskBusyQueue.empty())
    {
        auto task = m_taskBusyQueue.front();

        if (!task->basBusy() && task->m_runms <= nms)
        {
            // 任务完成：转移到空闲队列
            m_taskIdleMap.insert({task->getId(), task});

            // 将节点从繁忙队列直接拼接到空闲队列末尾
            m_taskIdleQueue.splice(m_taskIdleQueue.end(), m_taskBusyQueue, m_taskBusyQueue.begin());

            // 批量转移返回任务（假设 m_retTask 也是 std::list）
            if (!task->m_retTask.empty())
            {
                m_taskDataQueue.splice(m_taskDataQueue.end(), task->m_retTask);
            }
        }
        else
        {
            // 任务未完成：转移到临时队列
            m_taskTempQueue.splice(m_taskTempQueue.end(), m_taskBusyQueue, m_taskBusyQueue.begin());
        }
    }

    // 还原繁忙队列
    m_taskBusyQueue.splice(m_taskBusyQueue.end(), m_taskTempQueue);

    // 收集空闲线程
    std::queue<cgzCThread *> todothreads;
    if (!m_taskDataQueue.empty())
    {
        for (int i = 0; i < m_nThreadNum; i++)
        {
            auto work = m_pThreadList[i];
            if (work->hasIdle())
            {
                todothreads.push(work);
            }
        }
    }

    // 分配任务
    std::set<int> setTask;
    std::queue<cgzITask *> todoTasks;
    auto dataIt = m_taskDataQueue.begin();
    while (dataIt != m_taskDataQueue.end() && todoTasks.size() < todothreads.size())
    {
        auto task = *dataIt;
        auto idleIt = m_taskIdleMap.find(task->target);
        if (idleIt != m_taskIdleMap.end())
        {

            // 第一次放入该服务任务
            auto otask = idleIt->second;
            if (setTask.insert(idleIt->first).second)
            {
                todoTasks.push(otask);
            }

            // 该服务不接受新任务
            if (otask->addTask(task))
            {
                // 放入服务
                dataIt = m_taskDataQueue.erase(dataIt);
                continue;
            }
        }

        // 迭代下一个
        ++dataIt;
    }

    // 唤醒执行任务
    setTask.clear();
    std::map<cgzCThread *, cgzITask *> workthreads;
    while (!todoTasks.empty() && !todothreads.empty())
    {
        auto task = todoTasks.front();
        todoTasks.pop();
        task->setBusy();
        auto work = todothreads.front();
        todothreads.pop();
        m_taskIdleMap.erase(task->getId()); // 直接移除
        setTask.insert(task->getId());
        workthreads[work] = task;
        m_taskBusyQueue.push_back(task);
    }

    // 删除空闲队列中-将要繁忙的任务
    auto it = m_taskIdleQueue.begin();
    while (it != m_taskIdleQueue.end())
    {
        auto task = *it;
        if (setTask.count(task->getId()) > 0)
        {
            it = m_taskIdleQueue.erase(it);
        }
        else
        {
            if (m_bShutdown)
            {
                task->shutdown();
            }
            it++;
        }
    }
    // 还原空闲队列
    m_taskIdleQueue.splice(m_taskIdleQueue.end(), m_taskTempQueue);

    // 唤醒空闲线程
    for (auto &work : workthreads)
    {
        work.first->setbusy(work.second);
        work.first->wakeUp();
    }

    // 结束逻辑检查
    while (m_bShutdown && m_bRunning && m_taskDataQueue.empty())
    {

        // 遍历检查是否只有定时
        for (auto t : m_taskDataQueue)
        {
            if (t->target != 1)
            {
                return;
            }
        }

        // 遍历检查是否可以结束

        for (auto t : m_taskIdleQueue)
        {
            if (!t->mabystop())
            {
                return;
            }
        }

        // 确定已经可以结束服务
        for (auto t : m_taskIdleQueue)
        {
            t->realstop();
        }
        m_bRunning = false;
        std::cerr << "stop all thread" << std::endl;
        break;
    }
}
